package com.hopu.user.era;

import com.alibaba.fastjson.JSONObject;
import com.hopu.bean.era.TotalViewBean;
import com.hopu.util.HBaseUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.io.IOException;
import java.util.List;

public class TotalView {

    public static void main(String[] args) {
        transform(loadData());
    }

    //1.加载日志文件数据
    public static JavaRDD<String> loadData() {
        SparkConf conf = new SparkConf().setAppName("totalView").setMaster("local");
        JavaSparkContext context = new JavaSparkContext(conf);
        JavaRDD<String> rdd = context.textFile("D:/word/user_session.log");
        return rdd;

    }

    //2.转换，统计符合需要的数据
    public static void transform(JavaRDD<String> rdd) {

        JavaPairRDD<TotalViewBean, Integer> pairRDD = rdd.mapToPair(t -> {
            JSONObject jsonObject = (JSONObject) JSONObject.parse(t);
            String era = jsonObject.getJSONObject("birthday").getString("year").substring(2, 3) + "0";
            String event_type = jsonObject.getString("event_type");
            return new Tuple2<>(new TotalViewBean(era, event_type, 1), 1);
        });

        //过滤出每个年代的RDD
        String[] eras = {"50", "60", "70", "80", "90", "00", "10"};
//        filterByEra(pairRDD, "50");
        for (int i = 0; i < eras.length; i++) {
            filterByEra(pairRDD, eras[i]);
        }
    }

    public static void filterByEra(JavaPairRDD<TotalViewBean, Integer> pairRDD, String era) {
        JavaPairRDD<TotalViewBean, Integer> filter = pairRDD.filter(t -> era.equals(t._1.getEra())
        && "view".equals(t._1.getEvent_type()));
        //50 分组进行统计 key [1,1,1]
        JavaPairRDD<TotalViewBean, Iterable<Integer>> group = filter.groupByKey();

        JavaRDD<TotalViewBean> map = group.map(t -> {
            TotalViewBean key = t._1;
            int count = 0;
            for (int v : t._2) {
                count += v;
            }
            key.setCount(count);
            return key;
        });
        List<TotalViewBean> take = map.take(1);

//        group.foreach(t -> System.out.println(t));
        System.out.println(take.get(0));
        TotalViewBean bean = take.get(0);
        //3.把统计分析出的结果写入到HBase
        try {
            if (!HBaseUtils.tableExists("era")) {
                HBaseUtils.createTable("era", "totalView", "totalCart",
                        "totalParchase", "totalOperate", "top3cate", "top10shop");
            }
            String[] columns = {"era", "event_type", "count"};
            String[] values = {bean.getEra(), bean.getEvent_type(), bean.getCount()+""};
            HBaseUtils.putOneRowToHbase("era", bean.getEra(), "totalView", columns, values);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


}
